
//Copyright 1997-2009 Syrinx Development, Inc.
//This file is part of the Syrinx Web Application Framework (SWAF).
// == BEGIN LICENSE ==
//
// Licensed under the terms of any of the following licenses at your
// choice:
//
//  - GNU General Public License Version 3 or later (the "GPL")
//    http://www.gnu.org/licenses/gpl.html
//
//  - GNU Lesser General Public License Version 3 or later (the "LGPL")
//    http://www.gnu.org/licenses/lgpl.html
//
//  - Mozilla Public License Version 1.1 or later (the "MPL")
//    http://www.mozilla.org/MPL/MPL-1.1.html
//
// == END LICENSE ==
using System;
using System.Collections;
using System.Diagnostics;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;

using Swaf.Crypto;
using Swaf.Container;

namespace Swaf
{
	/// <summary>
	/// A class that inherits from the System.Net.Sockets.UdpClient in order to set advanced
	/// socket options so that multiple UdpClient objects can listen on the same IP address/port.
	/// This is important for the UdpServerStateManager given that multiple instances of it can be
	/// used on the same machine at the same time.</summary>
	public class AdvUdpClient : UdpClient
	{
		/// <summary>
		/// Intializes the UpdClient object for the given port on any IpAddress the
		/// code wants to use.</summary>
		public AdvUdpClient(int port)
		{
			Client.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, 1);
			Client.Bind(new IPEndPoint(IPAddress.Any, port));
		}
	}

	/// <summary>
	/// Implements the IServerStateManager by use UDP multicast technology.  </summary>
	/// <remarks>While the UDP multicast technology is not guarenteed to deliver
	/// its datagrams in order or even to deliver them at all this class manages these
	/// issues with a recieved messages IMap and will request datagrams to be resent
	/// based on missing ones.  Each message sent by the client will be broken down
	/// into at least 3 messages as a header and footer is always sent.  Data from the 
	/// client that is over 60000 bytes will be broken down into multiple 
	/// datagrams.  Each message has a header attached to it that includes a message id
	/// and which datagram it is out of the total number of datagrams that make up
	/// the message.
	/// 
	/// This class uses a thread to monitor received messages and if a message has
	/// not received all of its pieces and no datagram has come in for a set amount of
	/// time, the thread will send a resend message to the machine the origional message
	/// was recieved from.
	/// 
	/// The current implementation of this retry technology is not for mission critical
	/// situations.  It works well, but under extremely heavy load it is possible for every
	/// datagram from a message to be lost.  If that happens this class doesnt know datagrams
	/// were lost and wont ask for the message to be resent.  It works well when things
	/// needs to stay in sync and if missing messages are lost the application wont be
	/// totally broken from it.
	/// 
	/// This class is intended to be used by other classes to help them stay in sync with
	/// other instances of themselves.  Those classes can provide a "resync" type of message
	/// that will cause the objects to resync with eachother to help avoid situations where
	/// lost message cause failures.
	/// 
	/// Under observation, this class appears to work very well when the number of messages
	/// being sent in the conversation is not in the hundreds a second.  If the changes are 10
	/// or so a second or less, things work well.  UDP looses datagrams the most under heavy 
	/// load.
	/// 
	/// This class only supports one observer per message type and one
	/// global message observer. It will notify a specific message 
	/// observer and a global observer for the same message though.
	///
	/// This class also supports using the framework's encryption abstraction 
	/// to encrypt/decrypt the messages being used by this object on both 
	/// sends and receives.  However, it doesnt support being in an encrypt
	/// mode and receiving unencrypted methods.  If a message piece cannot be
	/// unencrypted it is assumed to be corrupted not unencrypted and will
	/// be asked to be sent again.
	/// 
	/// This class supports multiple conversations on a single IP address,
	/// but it is not recommended to do that because of the nature of 
	/// UDP reliability.  The more conversations on a single IP address means
	/// the more of a chance there is that collisions occur which will slow
	/// down performance due to resends and increase the possiblility that
	/// every datagram for a small message will be lost.
	/// 
	/// Using UDP multicast reliably is not trivial.  For more information
	/// about UDP multicast and building a reliable communication 
	/// stream from it see the online documentation for System.Net.Sockets.UdpClient
	/// and other online sources.
	/// 
	/// This class has a static shutdown method that shutdown every instance of
	/// the UdpServerStateManager.  This class does not attempt to register
	/// with the framework's shutdown event to trigger this and is up to
	/// another class to do it.</remarks>
	public class UdpServerStateManager : IServerStateManager
	{
		private static string s_resendBlockMsg = "______RSND______";
		private static int s_maxDatagramSize = 60000; //Real UDP max datagram size is 65k, but that includes header info.  This is safe.
		/// <summary>
		/// This static list of state manager is used to help shutdown the threads
		/// involved when the application is shutting down.</summary>
		private static IList s_allStateManagers = new ArrayList();

		protected AdvUdpClient m_multiCast = null;
		protected string m_cryptoKeyName;
		protected IPEndPoint m_multiCastGroupEndPoint;
		protected int m_portNum;
		protected int m_ttlDistance;

		protected IDictionary m_handlers = new Hashtable();
		protected processMessage m_globalHandler = null;
		protected ICryptoManager m_crypto = null;
		/// <summary>
		/// This is the binary formatter used when sending messages of type object
		/// that must be serialized into a byte array first.  This instance is created
		/// on first use so that if a object type is never sent then this instance
		/// will never be created.</summary>
		protected System.Runtime.Serialization.Formatters.Binary.BinaryFormatter m_binFormatter = null;

		/// <summary>
		/// The identity of the conversation, not the UdpServerStateManager
		/// instance.  When messages are received on the ip/port that dont have
		/// that id they are ignored.  That would mean there are multiple
		/// conversations on the same ip/port.</summary>
		protected string m_identity;
		protected INameAddressSupport m_pastSentMessages = new FlexiMap();
		protected IMap m_receivedMessages = new FlexiMap();

		/// <summary>
		/// Initializes the object to start listening to the given UDP multicast 
		/// group ip addess/port num for all machines within the given hops from 
		/// the machine this object is on.</summary>
		/// <remarks>
		/// This is the only constructor for this class which starts the threads to listen
		/// to the UPD socket and another thread to monitor the incoming messages in the
		/// m_receivedMessages dictionary.</remarks>
		/// <param name="identity">the conversation id this class is a member of.</param>
		/// <param name="multiCastGroupIP">The ip address of the upd multicast.  Must
		/// be within the defined multicast ip address range of 224.0.0.0 to 239.255.255.255.</param>
		/// <param name="portNum">The port num of the coversation</param>
		/// <param name="ttlDistance">Even though its called ttlDistance, its actually
		/// the number of network hops.  This number should be small, like 1 through 2 or 3.
		/// The bigger the number the larger the range of machines that can here the message
		/// and participate or clog up the ip/port for UDP failure.</param>
		/// <param name="cryptoKeyName">The valid framework crypto keyname
		/// to use.  If null, no encryption is used.  It is recommended to use
		/// an encryption key because ANYONE can easily see all the messages
		/// in the conversation because of the very unsecure nature of UDP
		/// multicast.</param>
		public UdpServerStateManager(string identity, string multiCastGroupIP, 
			int portNum, int ttlDistance, string cryptoKeyName)
		{
			lock (s_allStateManagers)
				s_allStateManagers.Add(this);

			m_identity = identity==null?"":identity;
			m_cryptoKeyName = cryptoKeyName;
			if (m_cryptoKeyName != null)
				m_crypto = Application.globalApp.appData.get("sysObjs-Crypto") as ICryptoManager;

			m_portNum = portNum;
			m_multiCastGroupEndPoint = new IPEndPoint(IPAddress.Parse(multiCastGroupIP), portNum);
			m_ttlDistance = ttlDistance;
			m_multiCast = new AdvUdpClient(portNum);
			m_multiCast.JoinMulticastGroup(m_multiCastGroupEndPoint.Address);

			Thread listenerThread =	new Thread(new ThreadStart(listener));
			listenerThread.ApartmentState = ApartmentState.MTA;
			listenerThread.Start();

			Thread blockCheckThread = new Thread(new ThreadStart(checkForMissingBlocks));
			blockCheckThread.ApartmentState = ApartmentState.MTA;
			blockCheckThread.Start();
		}

		/// <summary>
		/// registers the global handler, of which there can only be one.  Registering 
		/// a handler will over-write any exisiting global handler and they will not be
		/// called anymore.</summary>
		/// <param name="handler">The global observer</param>
		public void registerMessageHandler(processMessage handler)
		{
			m_globalHandler = handler;
		}

		/// <summary>
		/// Registers a handler for a specific type of message, of which there can only
		/// be one handler.  Registering a handler for the message type will
		/// over-write any exisiting handler for that message and that observer
		/// will no longer be called for incoming messages.</summary>
		/// <param name="msgType">The message type being observed.</param>
		/// <param name="handler">The observer to be registered.</param>
		public void registerMessageHandler(string msgType, processMessage handler)
		{
			m_handlers[msgType] = handler;
		}

		/// <summary>
		/// Removes the observer for the given message type.</summary>
		/// <param name="msgType">The message type to remove the 
		/// observer for.  If null the global observer will be
		/// removed.</param>
		public void unregisterMessageHandler(string msgType)
		{
			if (msgType == null)
				m_globalHandler = null;
			else if(m_handlers.Contains(msgType))
				m_handlers.Remove(msgType);
		}

		/// <summary>
		/// Will release the threads that are used by this instance.</summary>
		public void stop()
		{
			if (m_multiCast != null)
				lock (this)
					if (m_multiCast != null)
					{
						// Cleanly withdraw our membership from the multicast group
						m_multiCast.DropMulticastGroup(m_multiCastGroupEndPoint.Address);

						// Closing the underlying socket will cause UdpClient.Receive to throw
						m_multiCast.Close();
						m_multiCast = null;
						lock(s_allStateManagers)
							s_allStateManagers.Remove(this);
					}
		}

		/// <summary>
		/// Sends a string type message of the given message type.</summary>
		/// <param name="msgType">The message type of the given message.
		/// Must not be null.</param>
		/// <param name="message">The message contents. Must not be null.</param>
		public void sendMessage(string msgType, string message)
		{
			StringBuilder buff = new StringBuilder(m_identity.Length + message.Length + msgType.Length + 10);
			buff.Append(m_identity);
			buff.Append(":");
			buff.Append(msgType);
			buff.Append(":S");
			buff.Append(message);
			sendMessage(Encoding.UTF8.GetBytes(buff.ToString()));
		}

		public void sendMessage(string msgType, object anObject)
		{
			if (anObject is string)
				sendMessage(msgType, (string)anObject);
			else
			{
				System.IO.MemoryStream ms = new System.IO.MemoryStream();
				StringBuilder buff = new StringBuilder(msgType.Length + m_identity.Length + 10);
				buff.Append(m_identity).Append(":").Append(msgType).Append(":N");
				byte[] msgInfo = Encoding.UTF8.GetBytes(buff.ToString());
				ms.Write(msgInfo, 0, msgInfo.Length);
				BinFormatter.Serialize(ms, anObject);

				sendMessage(ms.ToArray());
			}
		}

		public void sendMessage(string msgType, byte[] message)
		{
			byte[] info = new byte[m_identity.Length + message.Length + msgType.Length + 3];
			StringBuilder buff = new StringBuilder(m_identity.Length + msgType.Length + 10);
			buff.Append(m_identity).Append(":").Append(msgType).Append(":B");
			Encoding.UTF8.GetBytes(buff.ToString(), 0, buff.Length, info, 0);
			Array.Copy(message, 0, info, 0, message.Length);
			sendMessage(info);
		}

		protected void sendMessage(byte[] msg)
		{
			if (m_crypto != null)
				msg = m_crypto.encrypt(m_cryptoKeyName, msg);
			
			string msgId = (Guid.NewGuid()).ToString();
			//We save the message in the m_pastMessages fleximap so that when requests come in for
			//blocks of a message we can turn to this to get it.
			m_pastSentMessages.put(msgId, msg, "expire in 1 min");

			int numChunks = calcNumChunks(msg);
			//Write header msg
			writeMessageBlock(msgId, msg, 0, numChunks+2);

			//Write msg contents
			for (int chunk = 1; chunk <= numChunks; ++chunk)
				writeMessageBlock(msgId, msg, chunk, numChunks + 2);

			//Write end msg.
			writeMessageBlock(msgId, msg, numChunks + 1,numChunks+2);
		}

		protected int calcNumChunks(byte[] msg)
		{
			int msgSize = msg.Length;
			int numChunks = (int)(msg.Length / s_maxDatagramSize);
			if (msg.Length % s_maxDatagramSize > 0)
				numChunks++;
			return numChunks;
		}

		protected void writeMessageBlock(string msgId, byte[] msg, int blockNum, int totalBlocks)
		{
			writeMessageBlock(msgId, msg, blockNum, totalBlocks, m_multiCastGroupEndPoint);
		}
		protected void writeMessageBlock(string msgId, byte[] msg, int blockNum, int totalBlocks, IPEndPoint dest)
		{
			System.IO.MemoryStream ms = new System.IO.MemoryStream();
			StringBuilder buff = new StringBuilder();
			buff.Append(msgId);
			buff.Append(":");
			buff.Append(blockNum);
			buff.Append(":");
			buff.Append(totalBlocks);
			buff.Append(":");

			byte[] block = null;

			if (blockNum == 0 || blockNum == totalBlocks -1)
			{
				block = new byte[buff.Length];
				Encoding.UTF8.GetBytes(buff.ToString(), 0, buff.Length, block, 0);
			}
			else
			{
				int dataSize = (int)((blockNum != totalBlocks-2) ? s_maxDatagramSize : Math.Abs((msg.Length - (s_maxDatagramSize * (totalBlocks - 3)))));
				block = new byte[buff.Length + dataSize];
				Encoding.UTF8.GetBytes(buff.ToString(), 0, buff.Length, block, 0);
				int msgIndex = ((blockNum-1)*s_maxDatagramSize);
				Array.Copy(msg, msgIndex, block, buff.Length, dataSize);
			}
			Debug.Assert(block != null);
			m_multiCast.Send(block, block.Length, dest);
		}

		public static void stopAll()
		{
			lock (s_allStateManagers)
			{
				foreach (UdpServerStateManager mgr in s_allStateManagers)
					mgr.stop();
				s_allStateManagers.Clear();
			}
		}

		protected System.Runtime.Serialization.Formatters.Binary.BinaryFormatter BinFormatter
		{
			get
			{
				if (m_binFormatter == null)
					lock (this)
						if (m_binFormatter == null)
							m_binFormatter =
								new System.Runtime.Serialization.Formatters.Binary.BinaryFormatter();
				return m_binFormatter;
			}
		}

		protected void checkForMissingBlocks()
		{
			while (true)
			{
				Thread.Sleep(250);
				lock (m_receivedMessages)
				{
					foreach(MessageBlocks msg in m_receivedMessages.Values)
						if (msg.Blocks.Count < msg.TotalBlocks && ((TimeSpan)(DateTime.Now - msg.LastMessage)).Seconds > 5)
						{
							msg.LastAttempt = DateTime.Now;
							if (((TimeSpan)(msg.LastAttempt - msg.LastMessage)).Seconds > 30)
							{
								//Dead message!!! 
								m_receivedMessages.Remove(msg.MessageId);
								Application.globalApp.Log.write("Server state manager id '{0}' throwing out message id '{1}' due to inability to retrieve all blocks",
									m_identity, msg.MessageId);
								break;
							}
							else
							{
								//Missing blocks for a while
								int nextBlock = 0;
								foreach (BlockInfo b in msg.Blocks.Values)
								{
									if (b.BlockNum != nextBlock)
									{
										for (int x = nextBlock; x < b.BlockNum; ++x)
										{
											//Send request for this missing block.
											int missingBlockNum = x;
											StringBuilder buff = new StringBuilder(1000);
											buff.Append(m_identity);
											buff.Append(":");
											buff.Append(s_resendBlockMsg);
											buff.Append(":S");
											buff.Append(msg.MessageId).Append(":");
											buff.Append(missingBlockNum).Append(":");
											byte[] block = new byte[buff.Length];
											Encoding.UTF8.GetBytes(buff.ToString(), 0, buff.Length, block, 0);
											m_multiCast.Send(block, block.Length, msg.SourceIP);
										}
									}
									nextBlock = b.BlockNum + 1;
								}
							}
						}
				}
			}
		}

		protected void listener()
		{
			// Loop forever (until underlying socket is closed, anyway)
			try
			{
				while (true)
				{
				whileLoop:
					try
					{
						// Wait for broadcast... will block until data recv'd, or underlying socket is closed
						IPEndPoint callerEndpoint = null;
						byte[] request = m_multiCast.Receive(ref callerEndpoint);
						lock (m_receivedMessages)
						{
							int x = 0;
							int index = 0;
							BlockInfo block = new BlockInfo();
							for (; x < request.Length; ++x)
							{
								if (request[x] == (byte)':')
								{
									block.MsgId = Encoding.UTF8.GetString(request, index, x - index);
									index = x += 1;
									break;
								}
							}
							for (; x < request.Length; ++x)
							{
								if (request[x] == (byte)':')
								{
									string nextInfo = Encoding.UTF8.GetString(request, index, x - index);
									if (nextInfo == s_resendBlockMsg)
									{
										if(block.MsgId == m_identity)
											processReceivedMessage(request,callerEndpoint);
										goto whileLoop;
									}
									else
									{
										block.BlockNum = int.Parse(nextInfo);
									}
									index = x += 1;
									break;
								}
							}
							for (; x < request.Length; ++x)
							{
								if (request[x] == (byte)':')
								{
									block.TotalBlocks = int.Parse(Encoding.UTF8.GetString(request, index, x - index));
									index = x += 1;
									break;
								}
							}
							block.blockData = new byte[request.Length - index];
							Array.Copy(request, index, block.blockData, 0, request.Length - index);
							MessageBlocks msgPieces = (MessageBlocks)m_receivedMessages.get(block.MsgId);
							if (msgPieces == null)
							{
								msgPieces = new MessageBlocks();
								msgPieces.MessageId = block.MsgId;
								msgPieces.TotalBlocks = block.TotalBlocks;
								msgPieces.SourceIP = callerEndpoint;
								m_receivedMessages.put(block.MsgId, msgPieces);
							}

							msgPieces.Blocks[block.BlockNum] = block;
							msgPieces.LastAttempt = msgPieces.LastMessage = DateTime.Now;
							if (msgPieces.Blocks.Count == block.TotalBlocks) //All blocks have arrived
							{
								m_receivedMessages.Remove(msgPieces.MessageId);
								processReceivedMessage(msgPieces);
							}
						}
					}
					catch (System.Net.Sockets.SocketException)
					{ } // expected (client got too impatient?)
				}
			}
			catch (System.ObjectDisposedException)
			{ } // expected
			catch (System.NullReferenceException)
			{ } // also expected?
		}

        protected class MessageBlocks
		{
			public DateTime LastMessage;
			public DateTime LastAttempt;
			public string MessageId;
			public int TotalBlocks;
			public SortedList Blocks = new SortedList();
			public IPEndPoint SourceIP;
		}

		protected class BlockInfo
		{
			public string MsgId = null;
			public int BlockNum = 0;
			public int TotalBlocks = 0;
			public byte[] blockData = null;
		}

		protected void processReceivedMessage(MessageBlocks msg)
		{
			int index = 0;
			//Combine messages
			byte[] request = null;
			int totalMsgLength = 0;
			foreach (BlockInfo block in msg.Blocks.Values)
				if (block.BlockNum != 0 && block.BlockNum != block.TotalBlocks + 1)
					totalMsgLength += block.blockData.Length;
			request = new byte[totalMsgLength];
			foreach (BlockInfo block in msg.Blocks.Values)
				if (block.BlockNum != 0 && block.BlockNum != block.TotalBlocks + 1)
				{
					Array.Copy(block.blockData, 0, request, index, block.blockData.Length);
					index += block.blockData.Length;
				}

			processReceivedMessage(request, msg.SourceIP);
		}
		protected void processReceivedMessage(byte[] request, IPEndPoint source)
		{

			if (m_crypto != null)
				request = m_crypto.decrypt(m_cryptoKeyName, request);

			string msgType = null;
			int x = 0;
			int index = 0;
			char type = 'x';
			string identity = null;
			for (; x < request.Length; ++x)
			{
				if (request[x] == (byte)':')
				{
					identity = Encoding.UTF8.GetString(request, 0, x);
					index = x += 1;
					break;
				}
			}
			if (identity == m_identity)
			{
				for (; x < request.Length; ++x)
				{
					if (request[x] == (byte)':')
					{
						msgType = Encoding.UTF8.GetString(request, index, x - index);
						type = (char)request[x + 1];
						index = x + 2;
						break;
					}
				}

				if (msgType != null)
				{
					processMessage handler = m_handlers[msgType] as processMessage;
					if (handler != null || m_globalHandler != null)
					{
						object data = null;
						switch (type)
						{
							case 'S':
								data = Encoding.UTF8.GetString(request, index, request.Length - index);
								break;
							case 'N':
								System.IO.MemoryStream responseStream = new System.IO.MemoryStream(request.Length);
								responseStream.Write(request, index, request.Length - index);
								responseStream.Position = 0;
								data = BinFormatter.Deserialize(responseStream);
								break;
							case 'B':
								byte[] msgData = new byte[request.Length - index];
								Array.Copy(request, index, msgData, 0, request.Length - index);
								data = msgData;
								break;
							default:
								Application.globalApp.Log.write("Unexpect format of '{0}' received by UdpServerStateManager from caller {1}",
									type, source);
								break;
						}
						if (msgType == s_resendBlockMsg)
						{
							resend((string)data,source);
						}
						else
						{
							if (handler != null)
								handler(msgType, data);
							if (m_globalHandler != null)
								m_globalHandler(msgType, data);
						}
					}
					else
						Application.globalApp.Log.write("No handler for message type {0} received by UdpServerStateManager from caller {1}",
							msgType, source);
				}
			}
		}

		protected void resend(string data, IPEndPoint source)
		{
			string[] info = data.Split(':');
			Debug.Assert(info.Length == 3);
			byte[] msg = (byte[])m_pastSentMessages.get(info[0]);
			if (msg != null)
			{
				int blockNumNeeded = int.Parse(info[1]);
				writeMessageBlock(info[0], msg, blockNumNeeded, calcNumChunks(msg)+2, source);
			}
		}
	}
}
